package com.intretech.tj.mqtt;

import com.intretech.tj.global.parent.VerticleParent;
import com.intretech.tj.mqtt.element.MQTTElement;
import com.intretech.tj.mqtt.element.WSElement;
import com.intretech.tj.mqtt.entity.Subscriber;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.smallrye.mutiny.Multi;
import io.vertx.core.Promise;
import io.vertx.mutiny.core.Vertx;
import io.vertx.mutiny.core.buffer.Buffer;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
import lombok.extern.slf4j.Slf4j;

import static com.intretech.tj.mqtt.OutLog.ElementStart;
import static com.intretech.tj.mqtt.OutLog.ServerStart;

@Slf4j
@ApplicationScoped
public class MQTTVerticle extends VerticleParent {
    private final OutLog outLog;

    private final MQTTElement MQTTElement;

    private final WSElement wsElement;

    private final MQTTHandler mqttHandler;

    @Inject
    public MQTTVerticle(OutLog outLog, MQTTElement MQTTElement, WSElement wsElement, MQTTHandler mqttHandler) {
        this.outLog = outLog;
        this.MQTTElement = MQTTElement;
        this.wsElement = wsElement;
        this.mqttHandler = mqttHandler;
    }

    @Override
    public void start(Promise<Void> startPromise) {
        log.info(outLog.getMessage(ServerStart));
        Multi.createFrom().items(MQTTElement, wsElement)
                .onItem().transform(verticle ->
                        Vertx.vertx().deployVerticle(verticle)
                ).subscribe().with(verticleUni -> verticleUni
                        .subscribe()
                        .with(id -> log.info(outLog.getMessage(ElementStart, id))));
    }

    /**
     * 向主题发布消息
     *
     * @param topic    主题
     * @param qosValue qos等级
     * @param message  消息
     */
    public void publish(String topic, int qosValue, String message) {
        mqttHandler.getTopicsForWildcard(topic).forEach((key, subscribers) -> {
            for (Subscriber subscriber : subscribers) {
                mqttHandler.publish(subscriber.getClientID(), qosValue, key, message);
            }
        });
    }

    /**
     * 声明保留消息
     */
    public void retain(String topic, int qosValue, String message) {
        mqttHandler.setRetain(topic, MqttQoS.valueOf(qosValue), message);
        mqttHandler.getTopicsForWildcard(topic).forEach((key, subscribers) -> {
            for (Subscriber subscriber : subscribers) {
                mqttHandler.retain(subscriber.getClientID(), MqttQoS.valueOf(qosValue), key, Buffer.buffer(message));
            }
        });
    }
}
